package com.binance.gateway.mq;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.binance.gateway.constant.GwConstants;
import com.binance.master.constant.MQConstant;
import com.binance.master.models.IpBlackListMsg;
import com.binance.master.utils.RedisCacheUtils;
import com.binance.master.utils.StringUtils;
import com.rabbitmq.client.Channel;
import lombok.extern.log4j.Log4j2;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.io.IOException;
import java.security.InvalidParameterException;
import java.util.HashSet;
import java.util.Set;

/**
 * Created by Fei.Huang on 2018/8/27.
 */
@Log4j2
@Service
public class Consumer {

    public static final long ONE_YEAR_EXPIRE_IN_SECOND = 60 * 60 * 24 * 365L;

    @RabbitListener(queues = MQConstant.IP_BLACKLIST_QUEUE)
    public void syncIpBlacklist(@Payload String blackListMsgStr, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
            Channel channel) throws IOException {

        try {
            log.info("blackListMsg:{}", blackListMsgStr);

            if (StringUtils.isBlank(blackListMsgStr)) {
                throw new InvalidParameterException("blackListMsgStr is null");
            }

            IpBlackListMsg blackListMsg =
                    JSONObject.toJavaObject(JSON.parseObject(blackListMsgStr), IpBlackListMsg.class);

            if (null == blackListMsg) {
                throw new InvalidParameterException("blackListMsg is null");
            }

            ipBlacklistHandler(blackListMsg);

            // All going well, consumer sending ack to broker
            channel.basicAck(deliveryTag, false);

        } catch (InvalidParameterException e) {

            log.warn("discard the message, exception:", e);
            // Reject & Not-ReQueue
            channel.basicReject(deliveryTag, false);

        } catch (Exception e) {

            log.error("syncIpBlacklist failed, exception:", e);
            // Reject but ReQueue
            channel.basicReject(deliveryTag, true);
        }
    }

    private void ipBlacklistHandler(IpBlackListMsg ipBlackListMsg) {

        Set<String> blackListIps = RedisCacheUtils.get(GwConstants.GW_IP_BLACKLIST_RDS_KEY, Set.class);
        if (CollectionUtils.isEmpty(blackListIps)) {
            blackListIps = new HashSet<>();
        }

        if (ipBlackListMsg.getType() == IpBlackListMsg.Type.ADD) {
            for (String ip : ipBlackListMsg.getIps()) {
                blackListIps.add(ip);
            }
            RedisCacheUtils.set(GwConstants.GW_IP_BLACKLIST_RDS_KEY, blackListIps, ONE_YEAR_EXPIRE_IN_SECOND);
        }

        if (ipBlackListMsg.getType() == IpBlackListMsg.Type.DELETE) {
            for (String ip : ipBlackListMsg.getIps()) {
                blackListIps.remove(ip);
            }
            RedisCacheUtils.set(GwConstants.GW_IP_BLACKLIST_RDS_KEY, blackListIps, ONE_YEAR_EXPIRE_IN_SECOND);
        }
    }
}
